package com.example.klinemqflink.job;

import com.example.klinemqflink.model.Kline;
import com.example.klinemqflink.utils.LocalTimeUtil;
import com.example.klinemqflink.utils.TimeCeilingUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * KlineSink is responsible for handling the output of Kline data.
 * It implements the SinkFunction interface to process and log Kline data.
 */
@Component
public class KlineSink implements SinkFunction<Kline> {
    private static final Logger logger = LoggerFactory.getLogger(KlineSink.class);

    /**
     * Invoke method processes each Kline record.
     * Currently, it logs the Kline data.
     *
     * @param kline   the Kline data to be processed
     * @param context the context in which the data is processed
     */
    @Override
    public void invoke(Kline kline, Context context) {
        // Here we would typically write to a database or message queue
        // For now, we'll just log the kline data
        long startTime = kline.getStartTime();
        long endTime = kline.getEndTime();
        endTime = TimeCeilingUtil.getCeilingTimestamp(endTime,kline.getFrequency());
        String startTimeStr = LocalTimeUtil.timestampToDateTime(startTime);
        kline.setStartTimeStr(startTimeStr);
        String endTimeStr = LocalTimeUtil.timestampToDateTime(endTime);
        kline.setEndTimeStr(endTimeStr);
        kline.setEndTime(endTime);
        logger.info("Processed Kline: {}", kline);
    }
}
